package com.intct.flink.state;

import com.intct.hbase.bean.WindowBean;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @author gufg
 * @since 2025-06-30 10:11
 */
public class KeyByValueStateDemo {
    public static void main(String[] args) throws Exception {
        // 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 设置并行度
        env.setParallelism(1);

        // 数据源
        DataStreamSource<String> sourceDS = env.socketTextStream("jd-node", 8888);

        // 转换操作
        SingleOutputStreamOperator<WindowBean> mapDS = sourceDS.map(source -> {
            String[] sources = source.split(",");
            return WindowBean.builder().name(sources[0]).ts(Long.valueOf(sources[1])).vc(Integer.valueOf(sources[2])).build();
        });

        // 分区
        KeyedStream<WindowBean, String> keyByDS = mapDS.keyBy(key -> key.getName());

        // 分区状态
        SingleOutputStreamOperator<WindowBean> valueStateDS = keyByDS.process(new KeyedProcessFunction<String, WindowBean, WindowBean>() {

            ValueState<Integer> valueState;

            @Override
            public void open(Configuration parameters) throws Exception {
                // 创建一个值状态的描述
                ValueStateDescriptor<Integer> valueStateDescriptor = new ValueStateDescriptor<>("vs1", Integer.class);
                // 创建值状态
                valueState = getRuntimeContext().getState(valueStateDescriptor);
            }

            @Override
            public void processElement(WindowBean value, KeyedProcessFunction<String, WindowBean, WindowBean>.Context ctx,
                                       Collector<WindowBean> out) throws Exception {
                // 获取状态值
                Integer windowBeanState = valueState.value();
                if (windowBeanState == null) {
                    windowBeanState = 0;
                };

                // Integer windowBeanState = valueState.value() == null ? 0 : valueState.value();

                // 业务处理逻辑
                /**
                 * WindwonBean
                 *  s1,1,1
                 *  s1,1,2
                 *  s1,1,3
                 */
                Integer valueVc = windowBeanState + value.getVc();

                // 累计后大于等于5
                if (valueVc >= 5) {
                    // 写出数据
                    out.collect(WindowBean.builder().name(value.getName()).vc(valueVc).build());
                    valueState.clear();
                } else {
                    // 更新状态
                    valueState.update(valueVc);
                }
            }

        });

        // 输出
        valueStateDS.print();

        // 启动作业
        env.execute();
    }
}
